package com.gaussdb.process;


import com.gaussdb.config.MyConfig;
import com.gaussdb.source.GaussDBSourceFunction;
import com.gaussdb.source.GaussDBSqlSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class GaussToDoris {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        GaussDBSourceFunction<String> build = GaussDBSqlSource.<String>builder()
                .hostname(MyConfig.hostname)        // 服务器地址
                .port(8001)                         // 端口
                .username(MyConfig.gauss_username)  // 用户
                .password(MyConfig.gauss_password)  // 密码
                .database(MyConfig.gauss_database)  // 库名
                .tableList(MyConfig.white_table_list)    // 单表：public.tablename 全表：public.* 多表：public.tableA,public.tableB
                .slotName(MyConfig.gauss_replication_slot)  // 复制槽
                .replicationMode(MyConfig.gauss_replication_mode)  // 逻辑复制槽模式： 1 创建逻辑复制槽 ，2 使用逻辑复制槽 ，3 删除逻辑复制槽
                .build();

        DataStreamSource<String> source = env.addSource(build);
        source.print();
//
//        SingleOutputStreamOperator<Tuple2<Integer, String>> map = source.map(new MapFunction<String, Tuple2<Integer, String>>() {
//            String before_id;
//            String before_full_name;
//
//            String after_id;
//            String after_full_name;
//
//
//            @Override
//            public Tuple2<Integer, String> map(String value) throws Exception {
//                // 生成json对象
//                JSONObject jsonObject = new JSONObject(value);
//                // dml类型
//                String op = jsonObject.getString("op_type");
//
//
//                // before对象
////                String columns_name = jsonObject.getString("columns_name");
////                String columns_type = jsonObject.getString("columns_type");
//                Object columnsVal = jsonObject.get("columns_val");
//                JSONArray jsonArray = new JSONArray(columnsVal.toString());
//                Integer id = jsonArray.getInt(0);
//                String full_name = jsonArray.getString(1);
//
//
////                System.out.println(columnsVal);
//
////                JSONArray jsonArray = new JSONArray(columns_val);
////                Tuple2<Integer, String> tuple2 = new Tuple2<>(Integer.valueOf(jsonArray.getString(0)), jsonArray.getString(1));
//
//                // after对象
////                Object old_keys_name = jsonObject.get("old_keys_name");
////                Object old_keys_type = jsonObject.get("old_keys_type");
////                Object old_keys_val = jsonObject.get("old_keys_val");
//
//                return Tuple2.of(id,full_name);
//            }
//        });
//
////        map.print();
//        map.addSink(new RichSinkFunction<Tuple2<Integer, String>>() {
//            Connection conn = null;
//            PreparedStatement preparedStatement = null;
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                super.open(parameters);
//                Class.forName("com.mysql.cj.jdbc.Driver");
//                conn = DriverManager.getConnection("jdbc:mysql://113.45.13.150:9030/", "root", "");
//                preparedStatement = conn.prepareStatement("insert into test.t2_sink values(?,?)");
//            }
//
//            @Override
//            public void close() throws Exception {
//                preparedStatement.close();
//                conn.close();
//                super.close();
//            }
//
//            @Override
//            public void invoke(Tuple2<Integer, String> value, Context context) throws Exception {
//                Integer id = value.f0;
//                String f1 = value.f1;
//
//                preparedStatement.setInt(1,id);
//                preparedStatement.setString(2,f1);
//                preparedStatement.execute();
//            }
//        });


        env.execute();
    }
}
